package com.jhl.seckill.config;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.jhl.seckill.entity.SeckillOrder;
import com.jhl.seckill.entity.User;
import com.jhl.seckill.service.GoodsMQService;
import com.jhl.seckill.service.SeckillService;
import com.jhl.seckill.vo.GoodsVo;
import com.jhl.seckill.vo.RetrySendMsg;
import com.jhl.seckill.vo.SeckillMsg;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Component
@Slf4j
public class MyAckReceiver implements ChannelAwareMessageListener {
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    GoodsMQService goodsMQService;

    @Autowired
    private SeckillService seckillService;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        log.info("执行onMessage方法");
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        log.info("消息id=" + deliveryTag);

        String msg = message.toString();
        System.out.println(msg);
        String[] msgArray = msg.split("'");//可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
        Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(), 3);


        try {
//            模拟消费者异常
//            int val=1/0;
            log.info("消费的主题消息来自：" + message.getMessageProperties().getConsumerQueue());
            /**
             * 手动ACK
             * 第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
             */
            channel.basicAck(deliveryTag, true);

        } catch (Exception e) {
//            拒绝消息，重新发送
            channel.basicReject(deliveryTag, false);

            String messageId = String.valueOf(UUID.randomUUID());
            String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            Map<String, Object> sendData = new HashMap<>();
            sendData.put("messageId", messageId);
            sendData.put("messageData", msg);
            sendData.put("createTime", createTime);

            resendMessage("seckillExchange", "seckill.#", sendData, message);
        }

        String messageData = msgMap.get("messageData");
        SeckillMsg seckillMsg = JSONUtil.toBean(messageData, SeckillMsg.class);

        Long goodsId = seckillMsg.getGoodsId();
        User user = seckillMsg.getUser();
        GoodsVo goodsVo = goodsMQService.findGoodsVobyGoodsId(goodsId);
        if (goodsVo.getStockCount() < 1) {
            return;
        }
//        判断是否重复抢购
        SeckillOrder seckillOrder = (SeckillOrder) redisTemplate.opsForValue().get("order:" + user.getId() + ":" + goodsId);
        if (seckillOrder != null) {
            return;
        }
//        正式执行秒杀逻辑
        log.info("消费者正式执行秒杀逻辑");
        seckillService.seckill(user, goodsVo);

    }


    public void resendMessage(String exchange, String routingKey, Object obj, Message message) throws InterruptedException {

        log.info("执行resendMessage方法");

//        获取消息id
        String messageId = message.getMessageProperties().getMessageId();
        if (StrUtil.isNotBlank(messageId) && Convert.toInt(redisTemplate.opsForValue().get(messageId)) > 3) {
            log.info("===========>>>重试三次失败后依然失败存库人工处理" + obj.toString());
            RetrySendMsg retrySendMsg = new RetrySendMsg();
            retrySendMsg.setExchange(message.getMessageProperties().getReceivedExchange());
            retrySendMsg.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
            retrySendMsg.setMsgBody(JSONUtil.toJsonStr(obj));
            retrySendMsg.setClassName(ClassUtil.getClassName(obj, false));
            //TODO insert方法
            return;
        }
//        间隔
        Thread.sleep(500);
        rabbitTemplate.convertAndSend(exchange, routingKey, obj, msg -> {
            MessageProperties messageProperties = msg.getMessageProperties();
            //设置消息ID
            String new_messageId = Convert.toStr(UUID.randomUUID());
            if (StrUtil.isBlank(messageId)) {
                messageProperties.setMessageId(new_messageId);
                messageProperties.setExpiration(Convert.toStr(RabbitMQTopicConfig.EXPIRATION));
//                缓存消息次数置为1
                redisTemplate.opsForValue().set(new_messageId, "1");
            } else {
                messageProperties.setMessageId(messageId);
                messageProperties.setExpiration(Convert.toStr(RabbitMQTopicConfig.EXPIRATION));
                redisTemplate.opsForValue().set(messageId, Convert.toInt(redisTemplate.opsForValue().get(messageId)) + 1 + "");
            }
            return msg;
        });
        log.info("再次发送消息");
    }

    //{key=value,key=value,key=value} 格式转换成map
    private Map<String, String> mapStringToMap(String str, int entryNum) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",", entryNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}
